use crate::cache::IndexChange;
use crate::keys::VddKey;
use crate::record_rule_db::impl_rdb::{delete_rule, select_rules};
use crate::{
    cache, delete_by_key_head, field_db, load_by_key_head, log_db, model_db, run_by_store,
    sqlx_run_by_store, CF_FV, ROCKSDB,
};
use rust_rocksdb::WriteBatchWithTransaction;
use serde_json::json;
use std::any::type_name;
use std::sync::Arc;
use visit_dd_core::i18n_text::{E_ARG_BAD, E_ROCKSDB};
use visit_dd_core::model::DataStoreEngine;
use visit_dd_core::proj::{VddLogPart, APP_CONFIG};
use visit_dd_core::query::VddValueEvaluation;
use visit_dd_core::record::{
    RecordConsume, RecordRule, RecordRuleDetail, RR_CONSUME, RR_EVA, RR_UNIQUE,
};
use visit_dd_core::{alog, i18n, search_type_name, Ares, FieldId, RecordId, UserId};

pub trait RecordRuleDbTrait {
    fn get_display(&self) -> impl std::future::Future<Output = String> + Send;
}

impl RecordRuleDbTrait for VddValueEvaluation {
    async fn get_display(&self) -> String {
        let field = field_db::f_load(&self.get_field_id()).await;
        format!(
            "{} {} {}",
            field.label,
            field.field_type.get_type_name(),
            self.get_params_display()
        )
    }
}

impl RecordRuleDbTrait for RecordRule {
    async fn get_display(&self) -> String {
        match self.detail.as_ref() {
            RecordRuleDetail::RrUnique(field_ids) => {
                let mut display_part = String::from("包含字段: ");
                let fields = field_db::f_select_by_ids(field_ids).await;
                for f in fields.iter() {
                    display_part = format!("{}{},", display_part, f.label);
                }
                format!(
                    "{} {}",
                    search_type_name(type_name::<RecordRuleDetail>(), RR_UNIQUE),
                    display_part
                )
            }
            RecordRuleDetail::RrConsumeTimes { consumer_m_id, max } => {
                let model = model_db::m_load(*consumer_m_id).await;
                let mut display_part = format!("消费的模型:{}", model.name);
                display_part = format!("{} 所有字段总消费", display_part);
                if *max > 0 {
                    display_part = format!("{} 最多使用 {} 次", display_part, max);
                }
                format!(
                    "{} {}",
                    search_type_name(type_name::<RecordRuleDetail>(), RR_CONSUME),
                    display_part
                )
            }
            RecordRuleDetail::RrFitEvaluation(vve) => {
                format!(
                    "{} {}",
                    search_type_name(type_name::<RecordRuleDetail>(), RR_EVA),
                    vve.get_display().await
                )
            }
        }
    }
}

pub async fn rr_select_in_model(model_id: u64) -> Vec<Arc<RecordRule>> {
    run_by_store!(
        impl_rdb::select_rules,
        impl_sqlx::rr_select_in_model,
        model_id
    )
}

/// 查找所有被某个模型消费的模型所包含的消费限制规则
pub async fn rr_select_consume_by_model(model_id: u64) -> Vec<Arc<RecordRule>> {
    run_by_store!(
        impl_rdb::rr_select_consume_by_model,
        impl_sqlx::rr_select_consume_by_model,
        model_id
    )
}

pub async fn rr_save(rr: &RecordRule, operator: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::save_rule(&mut batch, rr).await
    } else {
        sqlx_run_by_store!(impl_sqlx::rr_save, rr)?
    }
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator.clone(), "新建记录约束".into()),
            VddLogPart::LpModel(rr.model_id),
            VddLogPart::LpRr(rr.id),
        ),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn rr_delete_one(model_id: u64, rr_id: u64, operator: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        delete_rule(&mut batch, model_id, rr_id);
    } else {
        sqlx_run_by_store!(impl_sqlx::rr_delete_one, model_id, rr_id)?
    }
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator.clone(), "删除记录约束".into()),
            VddLogPart::LpModel(model_id),
            VddLogPart::LpRr(rr_id),
        ),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn count_rule(model_ids: &[u64]) -> Vec<usize> {
    if model_ids.is_empty() {
        return vec![];
    }
    run_by_store!(impl_rdb::count_rule, impl_sqlx::count_rule, model_ids)
}

/// 当字段被删除时,字段参与的记录规则也需要被删除,如果是唯一性约束规则,唯一性约束也同时需要被删除
pub(crate) fn on_delete_field(
    batch: &mut WriteBatchWithTransaction<true>,
    model_id: u64,
    field_id: &FieldId,
) {
    let rules = select_rules(model_id);
    for rule in rules {
        if rule.has_fid(field_id) {
            delete_rule(batch, model_id, rule.id);
        }
        // 如果是唯一性约束规则,唯一性约束也同时需要被删除
        if rule.rvrd_unique() {
            let key_head = VddKey::UniqueRecordValue(rule.id, "".into());
            delete_by_key_head(batch, &key_head);
        }
    }
}

/// 加载一条记录在消费者模型中的消费记录
pub(crate) fn rc_select(
    rec_id: RecordId,
    consume_m_id: u64,
    consume_f_id: Option<FieldId>,
) -> Vec<RecordConsume> {
    let cf_fv = ROCKSDB.cf_handle(CF_FV).unwrap();
    let mut r = if consume_m_id > 0 {
        // 指定了模型
        let key = VddKey::BoundRecordConsume(rec_id.clone(), consume_m_id).to_string();
        let v = ROCKSDB.get_cf(&cf_fv, &key).expect(&i18n!(E_ROCKSDB));
        if let Some(v) = v {
            serde_json::from_slice::<Vec<RecordConsume>>(v.as_slice()).expect(&i18n!(E_ARG_BAD))
        } else {
            Vec::<RecordConsume>::new()
        }
    } else {
        // 未指定模型，在所有模型中加载
        let key_head = VddKey::BoundRecordConsume(rec_id.clone(), 0);
        let mut sub_r = Vec::<RecordConsume>::new();
        load_by_key_head(&key_head).iter().for_each(|v| {
            let consumes =
                serde_json::from_slice::<Vec<RecordConsume>>(v).expect(&i18n!(E_ARG_BAD));
            consumes.iter().for_each(|consume| {
                sub_r.push(consume.clone());
            });
        });
        sub_r
    };
    if consume_f_id.is_some() {
        let c_f_id = consume_f_id.clone().unwrap();
        r.retain(|c| c.consumer_f_id == c_f_id);
    }
    r
}

/// 添加一条记录在消费者模型中的消费记录
pub(crate) fn rc_save(batch: &mut WriteBatchWithTransaction<true>, rec_consume: &RecordConsume) {
    let mut consumes = rc_select(
        rec_consume.record_id.clone(),
        rec_consume.consumer_m_id,
        None,
    );
    consumes.push(rec_consume.clone());
    let key = VddKey::BoundRecordConsume(rec_consume.record_id.clone(), rec_consume.consumer_m_id)
        .to_string();
    let cf_fv = ROCKSDB.cf_handle(CF_FV).unwrap();
    batch.put_cf(&cf_fv, key, json!(consumes).to_string());
}

mod impl_rdb {
    use crate::keys::VddKey;
    use crate::record_db::{select_records_all, UniKeyTraitExtern};
    use crate::{count_in_key_head, delete_by_key_head, load_by_key_head, CF_UNIQUE, ROCKSDB};
    use rust_rocksdb::WriteBatchWithTransaction;
    use serde_json::json;
    use std::sync::Arc;
    use visit_dd_core::record::{RecordRule, RecordRuleDetail};
    use visit_dd_core::EMPTY_RID;

    /// 查看一批模型的记录规则数目
    pub(super) fn count_rule(model_ids: &[u64]) -> Vec<usize> {
        let mut res = Vec::<usize>::with_capacity(model_ids.len());
        for model_id in model_ids {
            let kh = VddKey::BoundRecordRule(*model_id, 0);
            let count = count_in_key_head(&kh, |_| true);
            res.push(count);
        }
        res
    }

    /// 加载一个模型的所有值规则
    pub(super) fn select_rules(model_id: u64) -> Vec<Arc<RecordRule>> {
        let kh = VddKey::BoundRecordRule(model_id, 0);
        load_by_key_head(&kh)
            .iter()
            .map(|x| {
                let rule = serde_json::from_slice::<RecordRule>(x).unwrap();
                Arc::new(rule)
            })
            .collect::<Vec<Arc<RecordRule>>>()
    }

    pub(crate) fn delete_rule(
        batch: &mut WriteBatchWithTransaction<true>,
        model_id: u64,
        update_ts: u64,
    ) {
        let key = VddKey::BoundRecordRule(model_id, update_ts).to_string();
        batch.delete(key);
        //  删除此规则对应的值hash
        let key_head = VddKey::UniqueRecordValue(update_ts, "".into());
        delete_by_key_head(batch, &key_head)
    }

    pub(crate) async fn save_rule(batch: &mut WriteBatchWithTransaction<true>, rule: &RecordRule) {
        let vk = VddKey::BoundRecordRule(rule.model_id, rule.id);
        batch.put_cf(
            vk.db_name().unwrap(),
            vk.to_string(),
            json!(rule).to_string(),
        );
        // 为现有的记录创建对应的值hash
        match rule.detail.as_ref() {
            RecordRuleDetail::RrUnique(..) => {
                select_records_all(rule.model_id)
                    .await
                    .iter()
                    .for_each(|record| {
                        let cf_u = ROCKSDB.cf_handle(CF_UNIQUE).unwrap();
                        let unique_key = record.get_unique_key_by_fids(rule);
                        batch.put_cf(&cf_u, &unique_key, &record.meta.id.0.to_le_bytes());
                    });
            }
            _ => {}
        }
    }

    /// 查找所有被某个模型消费的模型所包含的消费限制规则
    pub fn rr_select_consume_by_model(model_id: u64) -> Vec<Arc<RecordRule>> {
        let kh = VddKey::BoundRecordConsume(EMPTY_RID, 0);
        let mut res = Vec::<Arc<RecordRule>>::new();
        load_by_key_head(&kh).iter().for_each(|x| {
            let rule = serde_json::from_slice::<RecordRule>(x).unwrap();
            if rule.consume_by_model(model_id) {
                res.push(Arc::new(rule));
            }
        });
        res
    }
}

mod impl_sqlx {
    use crate::keys::VddKey;
    use crate::sqlx::SqlxDefineTrait;
    use std::sync::Arc;
    use visit_dd_core::record::RecordRule;
    use visit_dd_core::Ares;

    pub(super) async fn count_rule(pool: &impl SqlxDefineTrait, model_ids: &[u64]) -> Vec<usize> {
        let bts = model_ids
            .iter()
            .map(|x| VddKey::BoundRecordRule(*x, 0))
            .collect::<Vec<VddKey>>();
        let count_map = pool.count_bounds_grouped(&bts).await;
        model_ids
            .iter()
            .map(|x| match count_map.get(x) {
                Some(count) => *count,
                None => 0,
            })
            .collect::<Vec<usize>>()
    }

    pub(super) async fn rr_select_in_model(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
    ) -> Vec<Arc<RecordRule>> {
        let sql_r = pool
            .select_bounds(VddKey::BoundRecordRule(model_id, 0))
            .await;
        sql_r
            .iter()
            .map(|x| {
                let rr = serde_json::from_str::<RecordRule>(&x).unwrap();
                Arc::new(rr)
            })
            .collect::<Vec<Arc<RecordRule>>>()
    }

    pub(super) async fn rr_save(pool: &impl SqlxDefineTrait, rr: &RecordRule) -> Ares {
        pool.save_bound(
            VddKey::BoundRecordRule(rr.model_id, rr.id),
            &serde_json::to_string(&rr).unwrap(),
        )
        .await
    }

    pub(super) async fn rr_delete_one(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        rr_id: u64,
    ) -> Ares {
        pool.delete_bounds(VddKey::BoundRecordRule(model_id, rr_id))
            .await
    }

    /// 查找所有被某个模型消费的模型所包含的消费限制规则
    pub(super) async fn rr_select_consume_by_model(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
    ) -> Vec<Arc<RecordRule>> {
        let mut res = Vec::<Arc<RecordRule>>::new();
        let sql_r = pool.select_bounds(VddKey::BoundRecordRule(0, 0)).await;
        sql_r.iter().for_each(|x| {
            let rr = serde_json::from_str::<RecordRule>(&x).unwrap();
            if rr.consume_by_model(model_id) {
                res.push(Arc::new(rr));
            }
        });
        res
    }
}
